package spring.cloud.tasks.tasks_manager.schedule;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.tasks_manager.listener.TaskAndExecutorChangeListenerManager;
import spring.cloud.tasks.tasks_manager.service.TaskInstanceService;
import spring.cloud.tasks.tasks_manager.taskInstanceManager.BatchTaskTaskInstanceManager;
import spring.cloud.tasks.tasks_manager.taskInstanceManager.StateFullSimpleTaskTaskInstanceManager;
import spring.cloud.tasks.tasks_manager.taskInstanceManager.StateLessSimpleTaskTaskInstanceManager;
import spring.cloud.tasks.tasks_manager.task_item_assign.TaskItemAssign;
import spring.cloud.tasks.tasks_manager.task_item_assign.TaskItemContainer;
import spring.cloud.tasks.tasks_manager.waitingPreAssignTasks.SlidingTimeTaskBucketQueueService;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
@Setter
@Getter
public class ScheduleManager {
    public String nameSpace;
    public Curator curator;
    public TaskIdListContext taskIdListContext = new TaskIdListContext();
    public TaskConfigurationManager taskConfigurationManager;
    public ScheduledExecutorService taskConfigurationManagerScheduledExecutorService;
    //
    public TaskItemContainer taskItemContainer = new TaskItemContainer();


    public TaskFireDateTimeAndPreAssignDateTimeManager taskFireDateTimeAndPreAssignDateTimeManager;

    //
    public ScheduledExecutorService fireDateTimeAndPreAssignDateTimeManagerScheduledExecutorService;
    //
    public TaskInstanceRegisterManager taskInstanceRegisterManager;
    public TaskInstanceStateManager taskInstanceStateManager;
    //
    public ConcurrentHashMap<String, TaskInstanceInfo> managedTaskIdAndTaskInstanceIdToTaskInstanceInfo = new ConcurrentHashMap<String, TaskInstanceInfo>();
    public ConcurrentHashMap<String, TaskInstanceInfo> unManagedTaskIdAndTaskInstanceIdToTaskInstanceInfo = new ConcurrentHashMap<String, TaskInstanceInfo>();
    //
    public StateFullSimpleTaskTaskInstanceManager stateFullSimpleTaskTaskInstanceManager = new StateFullSimpleTaskTaskInstanceManager();
    public StateLessSimpleTaskTaskInstanceManager stateLessSimpleTaskTaskInstanceManager = new StateLessSimpleTaskTaskInstanceManager();
    public BatchTaskTaskInstanceManager batchTaskTaskInstanceManager = new BatchTaskTaskInstanceManager();
    public ScheduledExecutorService taskInstanceRegisterManagerScheduledExecutorService;
    public ScheduledExecutorService taskInstanceStateManagerScheduledExecutorService;

    public SlidingTimeTaskBucketQueueService slidingTimeTaskBucketQueueService = new SlidingTimeTaskBucketQueueService();
    //
    public ScheduledExecutorService taskItemAssignScheduledExecutorService;
    public Lock taskItemAssignLock = new ReentrantLock();
    public TaskAndExecutorChangeListenerManager taskAndExecutorChangeListenerManager;
    private AtomicBoolean needTaskItemReAssign = new AtomicBoolean(false);

    public boolean getNeedTaskItemReAssign() {
        return needTaskItemReAssign.get();
    }

    public void setNeedTaskItemReAssign(boolean needTaskItemReAssign) {
        this.needTaskItemReAssign.set(needTaskItemReAssign);
    }

    //
    private TaskInstanceService taskInstanceService;

    public ScheduleManager(String nameSpace, Curator curator) {
        this.nameSpace = nameSpace;
        this.curator = curator;

    }

    public void start() {
        if (taskConfigurationManagerScheduledExecutorService != null) {
            taskConfigurationManagerScheduledExecutorService.shutdownNow();
        }
        if (fireDateTimeAndPreAssignDateTimeManagerScheduledExecutorService != null) {
            fireDateTimeAndPreAssignDateTimeManagerScheduledExecutorService.shutdownNow();
        }
        if (taskInstanceRegisterManagerScheduledExecutorService != null) {
            taskInstanceRegisterManagerScheduledExecutorService.shutdownNow();
        }
        if (taskItemAssignScheduledExecutorService != null) {
            taskItemAssignScheduledExecutorService.shutdownNow();
        }
        taskConfigurationManagerScheduledExecutorService = buildTaskConfigurationManagerScheduledExecutorService();
        fireDateTimeAndPreAssignDateTimeManagerScheduledExecutorService = buildFireDateTimeAndPreAssignDateTimeManagerScheduledExecutorService();
        taskInstanceRegisterManagerScheduledExecutorService = buildTaskInstanceManagerScheduledExecutorService();
        taskInstanceStateManagerScheduledExecutorService = buildTaskInstanceStateManagerScheduledExecutorService();
        taskItemAssignScheduledExecutorService = buildTaskItemAssignScheduledExecutorService();
        //
        taskAndExecutorChangeListenerManager = new TaskAndExecutorChangeListenerManager(this);

    }

    public void stop() {
        if (fireDateTimeAndPreAssignDateTimeManagerScheduledExecutorService != null) {
            fireDateTimeAndPreAssignDateTimeManagerScheduledExecutorService.shutdownNow();
        }
        if (taskInstanceRegisterManagerScheduledExecutorService != null) {
            taskInstanceRegisterManagerScheduledExecutorService.shutdownNow();
        }
        if (taskInstanceStateManagerScheduledExecutorService != null) {
            taskInstanceStateManagerScheduledExecutorService.shutdownNow();
        }
        if (taskItemAssignScheduledExecutorService != null) {
            taskItemAssignScheduledExecutorService.shutdownNow();
        }
        if (taskConfigurationManagerScheduledExecutorService != null) {
            taskConfigurationManagerScheduledExecutorService.shutdownNow();
        }
    }


    public ScheduledExecutorService buildFireDateTimeAndPreAssignDateTimeManagerScheduledExecutorService() {

        taskFireDateTimeAndPreAssignDateTimeManager = new TaskFireDateTimeAndPreAssignDateTimeManager(this, nameSpace, curator);

        ScheduledExecutorService scheduledExecutorService = newSingleThreadExecutor(nameSpace);
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    //创建实例的时候进行更新
                    //taskFireDateTimeAndPreAssignDateTimeManager.refresh();
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        }, 15, 15, TimeUnit.SECONDS);
        return scheduledExecutorService;
    }

    public ScheduledExecutorService buildTaskInstanceManagerScheduledExecutorService() {

        taskInstanceRegisterManager = new TaskInstanceRegisterManager(this, nameSpace, curator);

        ScheduledExecutorService scheduledExecutorService = newSingleThreadExecutor(nameSpace);
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    taskFireDateTimeAndPreAssignDateTimeManager.refresh();
                    taskInstanceRegisterManager.refresh();
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        }, 15, 15, TimeUnit.SECONDS);
        return scheduledExecutorService;
    }

    public ScheduledExecutorService buildTaskInstanceStateManagerScheduledExecutorService() {

        taskInstanceStateManager = new TaskInstanceStateManager(this, nameSpace, curator);

        ScheduledExecutorService scheduledExecutorService = newSingleThreadExecutor(nameSpace);
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    Lock lock = taskInstanceStateManager.lock;
                    boolean tryLock = lock.tryLock(15, TimeUnit.SECONDS);//加锁失败 15秒后又会进行任务分配(30-15)
                    if (tryLock) {
                        try {
                            taskInstanceStateManager.refresh();
                        } catch (Throwable e) {
                            log.error(e.getMessage(), e);
                        } finally {
                            lock.unlock();
                        }

                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        }, 15, 15, TimeUnit.SECONDS);
        return scheduledExecutorService;
    }

    public ScheduledExecutorService buildTaskItemAssignScheduledExecutorService() {
        ScheduledExecutorService scheduledExecutorService = newSingleThreadExecutor(nameSpace);
        ScheduleManager scheduleManager = this;
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    Lock lock = scheduleManager.taskItemAssignLock;
                    boolean tryLock = lock.tryLock(15, TimeUnit.SECONDS);//加锁失败 15秒后又会进行任务分配(30-15)
                    if (tryLock) {
                        try {
                            TaskItemAssign taskItemAssign = new TaskItemAssign(scheduleManager);
                            taskItemAssign.taskItemAssign();
                        } catch (Throwable e) {
                            log.error(e.getMessage(), e);
                        } finally {
                            lock.unlock();
                        }

                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }


            }
        }, 30, 30, TimeUnit.SECONDS);
        return scheduledExecutorService;
    }

    public ScheduledExecutorService newSingleThreadExecutor(final String nameSpace) {
        return Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, nameSpace + "-" + r.getClass().getSimpleName());
            }
        });
    }

    public ScheduledExecutorService buildTaskConfigurationManagerScheduledExecutorService() {
        ScheduledExecutorService scheduledExecutorService = newSingleThreadExecutor(nameSpace);
        taskConfigurationManager = new TaskConfigurationManager(this);
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    taskConfigurationManager.refresh();
                } catch (Exception e) {
                    e.printStackTrace();
                }


            }
        }, 8, 8, TimeUnit.SECONDS);
        return scheduledExecutorService;
    }

}
